/*
 ***************************************************************************************
 * All rights Reserved, Designed By RTF
 * @Title:  AbstractMessageConsumer.java   
 * @Package com.rtf.repair.mq.common
 * @Description: 消费处理消息默认抽象实现，针对不同的mq 实现获取特殊信息以兼容不同的处理，如rocket
 * 				 具有tag 信息，其它mq可能会以其它形式存储（kafka放入header里）；针对不同mq默认实现
 * 				对应的消息体解析获取body和tag,如  onMessage(ConsumerRecord<String, byte[]> record)
 * 				解析之后执行onMessage(String tag, byte[] message)
 * 				
 * @author: sfl
 * @date:   2020-10-22 13:05:07   
 * @version V1.0 
 * @Copyright: 2020 FOSONG. All rights reserved. 
 * 注意：本内容仅限于公司内部使用，禁止外泄以及用于其他的商业目
 *  ---------------------------------------------------------------------------------- 
 * 文件修改记录
 *     文件版本：         修改人：             修改原因：
 ***************************************************************************************
 */
package com.rtf.framework.mq.common;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;

/**
 *  
 * @Description: 消费处理消息默认抽象实现，针对不同的mq 实现获取特殊信息以兼容不同的处理，如rocket
 * 				  具有tag 信息，其它mq可能会以其它形式存储（kafka放入header里）；针对不同mq默认实现
 * 				  对应的消息体解析获取body和tag,如  onMessage(ConsumerRecord<String, byte[]> record)
 * 				  解析之后执行onMessage(String tag, byte[] message)
 * @ClassName:  AbstractMessageConsumer
 * @author: sfl
 * @date:   2020-10-22 13:11:03
 * @since:  v1.0
 */
public abstract class AbstractMessageConsumer implements MessageConsumer {

	/* (non-Javadoc)
	 * @see com.rtf.repair.mq.common.MessageConsumer#onMessage(org.apache.kafka.clients.consumer.ConsumerRecord)
	 */
	@Override
	public boolean onMessage(ConsumerRecord<String, byte[]> record) {
		Headers headers = record.headers();
		Iterable<Header> heads = headers.headers(MessageConst.PROPERTY_TAGS);
		String tag = null;
		if (heads != null && heads.iterator().hasNext())
			tag = new String(heads.iterator().next().value());
		return this.onMessage(record.topic(), tag, record.timestamp(),record.value());
	}

	/* (non-Javadoc)
	 * @see com.rtf.repair.mq.common.MessageConsumer#onMessage(org.apache.rocketmq.common.message.MessageExt)
	 */
	@Override
	public boolean onMessage(MessageExt record) {
		return this.onMessage(record.getTopic(), record.getTags(),record.getBornTimestamp(), record.getBody());
	}

}
